package club.ruanx.movie.message;

import club.ruanx.exception.BaseException;
import club.ruanx.movie.adapter.SpiderMovieAdapter;
import club.ruanx.movie.adapter.SpiderPersonAdapter;
import club.ruanx.movie.common.property.SpiderProperties;
import club.ruanx.movie.service.MovieService;
import club.ruanx.movie.service.PersonService;
import club.ruanx.redis.entity.MessageData;
import club.ruanx.spider.common.message.SpiderMessageClient;
import club.ruanx.spider.entity.DoubanMovie;
import club.ruanx.spider.repository.DoubanMovieRepository;
import club.ruanx.util.CollectionUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.stereotype.Component;

import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author 阮胜
 * @date 2018/11/28 21:26
 */

@Component
@EnableBinding(SpiderMessageClient.class)
@Slf4j
public class RabbitMessageReceiver {

    private final DoubanMovieRepository doubanMovieRepository;
    private final SpiderMovieAdapter spiderMovieAdapter;
    private final SpiderPersonAdapter spiderPersonAdapter;
    private final MovieService movieService;
    private final PersonService personService;
    private final SpiderProperties spiderProperties;

    public RabbitMessageReceiver(DoubanMovieRepository doubanMovieRepository, SpiderMovieAdapter spiderMovieAdapter, SpiderPersonAdapter spiderPersonAdapter, MovieService movieService, PersonService personService, SpiderProperties spiderProperties) {
        this.doubanMovieRepository = doubanMovieRepository;
        this.spiderMovieAdapter = spiderMovieAdapter;
        this.spiderPersonAdapter = spiderPersonAdapter;
        this.movieService = movieService;
        this.personService = personService;
        this.spiderProperties = spiderProperties;
    }

    @StreamListener(SpiderMessageClient.INPUT)
    public void receive(MessageData spiderMessageData) {
        Page<DoubanMovie> moviePage = doubanMovieRepository.findAllBySynchronization(false, PageRequest.of(0, spiderProperties.getSyncRows()));
        List<DoubanMovie> movieList = moviePage.getContent();
        log.debug("开始同步爬虫数据,一共{}条数据", movieList.size());
        movieList.stream()
                .filter(doubanMovie -> {
                    if (movieService.existsBySpiderId(doubanMovie.getId())) {
                        doubanMovie.setSynchronization(true);
                        return false;
                    }
                    return true;
                })
                .map(doubanMovie -> {
                    try {
                        return spiderMovieAdapter.adapter(doubanMovie);
                    } catch (BaseException e) {
                        log.error(e.getMessage());
                        return null;
                    }
                }).filter(Objects::nonNull)
                .map(movie -> {
                    try {
                        movieService.save(movie);
                        log.info("新增电影:{}", movie.getName());
                        return movie.getSpiderId();
                    } catch (Exception e) {
                        e.printStackTrace();
                        return null;
                    }
                }).filter(Objects::nonNull)
                .forEach(spiderId -> movieList.stream()
                        .filter(doubanMovie -> doubanMovie.getId().equalsIgnoreCase(spiderId))
                        .findFirst()
                        .ifPresent(doubanMovie -> {
                            doubanMovie.setSynchronization(true);
                            doubanMovieRepository.save(doubanMovie);
                        }));


        movieList.stream().flatMap(doubanMovie -> {
            List<DoubanMovie.CastsBean> casts = doubanMovie.getCasts();
            List<DoubanMovie.DirectorsBean> directors = doubanMovie.getDirectors();
            Set<String> idSet = new HashSet<>();
            if (!CollectionUtils.isEmpty(casts)) {
                List<String> castIds = casts.stream()
                        .map(DoubanMovie.CastsBean::getId)
                        .collect(Collectors.toList());
                idSet.addAll(castIds);
            }
            if (!CollectionUtils.isEmpty(directors)) {
                List<String> directorIds = directors.stream()
                        .map(DoubanMovie.DirectorsBean::getId)
                        .collect(Collectors.toList());
                idSet.addAll(directorIds);
            }
            log.info("共获取{}条影人信息", idSet.size());
            return idSet.stream();
        }).filter(id -> !personService.existsBySpiderId(id))
                .map(id -> {
                    try {
                        return spiderPersonAdapter.adapter(id);
                    } catch (Exception e) {
                        log.error(e.getMessage());
                        return null;
                    }
                }).filter(Objects::nonNull)
                .forEach(person -> {
                    personService.save(person);
                    log.info("增加新的影人:{}", person.getName(), person);
                });
        log.info("来自爬虫消息队列的消息已成功处理.");
    }
}
